package com.os.ops.impl;

import cn.hutool.core.util.ObjectUtil;
import com.os.data.DataStore;
import com.os.config.RsConfig;
import com.os.constant.Sys;
import com.os.exception.Asserts;
import com.os.feign.DataService;
import com.os.feign.MetaDataService;
import com.os.model.DataSource;
import com.os.model.LocationMsg;
import com.os.model.MetaDataInfo;
import com.os.model.DataInfo;
import com.os.ops.Ops;
import com.os.ops.RsPutStream;
import com.os.result.Resp;
import com.os.result.ResultCode;
import com.os.rs.backblaze.ReedSolomon;
import com.os.task.*;
import com.os.util.ConvertMetaDataInfo;
import com.os.util.Hash;
import com.os.util.IoUtilPlus;
import feign.Response;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * @author 风亦未止
 * @date 2023/5/4 16:57
 */
@Service
@Slf4j
public class OpsImpl implements Ops {
    @Autowired
    private DataService dataService;
    @Autowired
    private MetaDataService metaDataService;
    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private MessagePostProcessor messagePostProcessor;
    @Autowired
    private RsConfig rsConfig;

    @SneakyThrows
    @Override
    public boolean putObj(MultipartFile file) {
        String name = file.getOriginalFilename();
        //查询元数据是否存在
        Resp respObj = metaDataService.get(name);
        MetaDataInfo metaDataGet = ConvertMetaDataInfo.convert(respObj);

        //计算hash
        String hash = null;
        try {
            hash = Hash.hash(file.getBytes());
        } catch (IOException e) {
            e.printStackTrace();
        }
        DataInfo dataInfo = new DataInfo();
        dataInfo.setHash(hash);

        //未上传过
        if (metaDataGet == null) {
            //创建元数据信息
            MetaDataInfo metaDataInfo = new MetaDataInfo();
            metaDataInfo.setVersion(0);
            metaDataInfo.setName(name);
            metaDataInfo.setSize((int) file.getSize());
            metaDataInfo.setHash(hash);
            metaDataInfo.setCreated(new Date());
            //定位hash看看是否需要写入数据
            List<LocationMsg> msgList = doLocate(dataInfo.getHash());
            if (msgList.size() < rsConfig.DATA_SHARDS) {
                //写入数据
                return feignPutObjAndMetaData(metaDataInfo, dataInfo, msgList, file);
            }
            return feignPutMetaData(metaDataInfo);
        }
        //对象数据可能存在了，对比hash
        if (hash.equals(metaDataGet.getHash())) {
            //是一样的就直接上传成功！
            //更新元数据
            metaDataGet.setName(name);
            metaDataGet.setCreated(new Date());
            metaDataGet.setHash(hash);
            metaDataGet.setVersion(metaDataGet.getVersion() + 1);
            metaDataGet.setSize((int) file.getSize());
            return feignPutMetaData(metaDataGet);
        }
        //哈希不一样，更新元数据,写入数据
        metaDataGet.setSize((int) file.getSize());
        metaDataGet.setName(name);
        metaDataGet.setCreated(new Date());
        metaDataGet.setVersion(metaDataGet.getVersion() + 1);
        metaDataGet.setHash(hash);
        //定位hash看看是否需要写入数据
        List<LocationMsg> msgList = doLocate(dataInfo.getHash());
        if (msgList.size() < rsConfig.DATA_SHARDS) {
            //写入数据
            return feignPutObjAndMetaData(metaDataGet, dataInfo, msgList, file);
        }
        return feignPutMetaData(metaDataGet);
    }

    @Override
    @SneakyThrows
    public boolean putStream(InputStream inputStream,String objName, String hash, int size) {
        RsPutStream rsPutStream = new RsPutStream(rsConfig, dataService, hash, size);
        //实际算出的hash值
        String culHash = IoUtilPlus.copyWithHash(inputStream, rsPutStream);
        System.out.println(culHash);
        if(culHash.equals(hash)){
            rsPutStream.commit(true);
            return true;
        }
        rsPutStream.commit(false);
        return false;
    }


    @Override
    public InputStream getObj(String name) {
        Resp resp = metaDataService.get(name);
        return getData(resp);
    }

    @Override
    public boolean delObj(String name) {
        Resp resp = metaDataService.get(name);
        MetaDataInfo metaDataInfo = ConvertMetaDataInfo.convert(resp);
        if (ObjectUtil.isNull(metaDataInfo) || metaDataInfo.getSize() == -1) {
            return false;
        }
        metaDataInfo.setSize(-1);
        metaDataInfo.setHash("");
        metaDataInfo.setVersion(metaDataInfo.getVersion() + 1);
        return feignPutMetaData(metaDataInfo);
    }

    @Override
    public InputStream getObjByVersion(String name, Integer version) {
        Resp resp = metaDataService.getByVersion(name, version);
        return getData(resp);
    }

    @Override
    public Resp getAllVersion(String name) {
        return metaDataService.getAllVersion(name);
    }

    /**
     * 根据获取的元数据去定位所有的数据分片，合并，获取数据流，回给客户端
     * @param resp
     * @return
     */
    @SneakyThrows
    private InputStream getData(Resp resp) {
        MetaDataInfo metaDataInfo = ConvertMetaDataInfo.convert(resp);
        if (ObjectUtil.isNull(metaDataInfo) || metaDataInfo.getSize() == -1) {
            Asserts.fail(ResultCode.NOTFOUND);
        }
        return locateData(metaDataInfo);
    }


    /**
     * 构建分片后上传
     * @param dataInfo    数据的信息-内含hash
     * @param locationMsg 已经上传的分片和节点的信息
     * @param file
     * @return
     */
    private boolean feignPutObjAndMetaData(MetaDataInfo metaData, DataInfo dataInfo, List<LocationMsg> locationMsg, MultipartFile file) {
        //构建分片后上传
        List<DataSource> dataSources = buildShared(file);
        PutShared putShared = new PutShared(dataService, dataInfo, locationMsg, dataSources, rsConfig);
        FutureTask<Resp> objTask = new FutureTask<>(putShared);
        PutMeta putMeta = new PutMeta(metaDataService, metaData);
        FutureTask<Resp> metaTask = new FutureTask<>(putMeta);
        return submit(metaTask, objTask);
    }


    /**
     * 提交线程任务
     * @param metaTask
     * @param objTask
     * @return
     */
    private boolean submit(FutureTask<Resp> metaTask, FutureTask<Resp> objTask) {
        threadPoolExecutor.submit(objTask);
        try {
            Resp resp = objTask.get();
            if (ObjectUtil.isNull(resp) || resp.getCode() != 200) {
                return false;
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            return false;
        }
        threadPoolExecutor.submit(metaTask);
        try {
            Resp resp = metaTask.get();
            if (ObjectUtil.isNotNull(resp) && resp.getCode() == 200) {
                return true;
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            return false;
        }
        return false;
    }

    /**
     * get对象 获取分片所在的数据节点 - 顺便进行数据修复
     *
     * @param metaDataInfo
     * @return
     */
    @SneakyThrows
    public InputStream locateData(MetaDataInfo metaDataInfo) {
        List<LocationMsg> ips = doLocate(metaDataInfo.getHash());
        return mergeAndRepair(ips, metaDataInfo);
    }

    /**
     * 开始定位分片
     * @param hash
     * @return
     */
    private List<LocationMsg> doLocate(String hash) {
        LocateData locateData = new LocateData(rabbitTemplate, messagePostProcessor, rsConfig, hash);
        FutureTask<List<LocationMsg>> dataNodeInfoFutureTask = new FutureTask<>(locateData);
        threadPoolExecutor.submit(dataNodeInfoFutureTask);
        List<LocationMsg> ip = null;
        try {
            ip = dataNodeInfoFutureTask.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            return ip;
        }
        return ip;
    }

    /**
     * 只做元数据的更新
     * @param metaDataInfo
     * @return
     */
    private boolean feignPutMetaData(MetaDataInfo metaDataInfo) {
        Resp re = metaDataService.put(metaDataInfo);
        if (!ObjectUtil.isNull(re) && re.getCode() == 200) {
            return true;
        }
        return false;
    }

    /**
     * 构建数据分片
     *
     * @param file
     * @return
     */
    @SneakyThrows
    private List<DataSource> buildShared(MultipartFile file) {
        int storedSize = (int) (file.getSize() + rsConfig.BYTES_IN_INT);
        int shardSize = (storedSize + rsConfig.DATA_SHARDS - 1) / rsConfig.DATA_SHARDS;
        int fileSize = (int) file.getSize();
        int bufferSize = shardSize * rsConfig.DATA_SHARDS;
        byte[] allBytes = new byte[bufferSize];
        ByteBuffer.wrap(allBytes).putInt(fileSize);
        InputStream inputStream = file.getInputStream();
        int bytesRead = inputStream.read(allBytes, rsConfig.BYTES_IN_INT, fileSize);
        if (bytesRead != fileSize) {
            throw new IOException("未读完所有的数据~");
        }
        inputStream.close();
        byte[][] shards = new byte[rsConfig.TOTAL_SHARDS][shardSize];
        for (int i = 0; i < rsConfig.DATA_SHARDS; i++) {
            System.arraycopy(allBytes, i * shardSize, shards[i], 0, shardSize);
        }
        rsConfig.reedSolomon.encodeParity(shards, 0, shardSize);
        List<DataSource> dataSources = new ArrayList<>();
        for (int i = 0; i < rsConfig.TOTAL_SHARDS; i++) {
            //这里就先把切片的索引放入originalFilename了..
            DataSource dataSource = new DataSource(i + "", shards[i]);
            dataSources.add(dataSource);
        }
        return dataSources;
    }

    /**
     * 合并数据片
     * @param ips
     * @param metaDataInfo
     * @return
     */
    @SneakyThrows
    private InputStream mergeAndRepair(List<LocationMsg> ips, MetaDataInfo metaDataInfo) {
        Map<Integer, String> locationInfo = new HashMap<>(ips.size());
        for (LocationMsg info : ips) {
            locationInfo.put(info.getIndex(), info.getAddress());
        }
        //小于数据分片数量，无法弥补 404
        if (ips.size() < rsConfig.DATA_SHARDS) {
            Asserts.fail(ResultCode.NOTFOUND);
        }
        //开始合并
        final byte[][] shards = new byte[rsConfig.TOTAL_SHARDS][];
        final boolean[] shardPresent = new boolean[rsConfig.TOTAL_SHARDS];
        int storedSize = metaDataInfo.getSize() + rsConfig.BYTES_IN_INT;
//        int storedSize = metaDataInfo.getSize();

        int shardSize = (storedSize + rsConfig.DATA_SHARDS - 1) / rsConfig.DATA_SHARDS;
        int shardCount = 0;
        for (int i = 0; i < rsConfig.TOTAL_SHARDS; i++) {
            if (locationInfo.get(i) != null) {
                URI uri = new URI(Sys.HTTP + locationInfo.get(i));
                Response response = dataService.get(uri, metaDataInfo.getHash());
                shardPresent[i] = true;
                shardCount += 1;
                shards[i] = new byte[shardSize];
                IOUtils.read(response.body().asInputStream(), shards[i], 0, shardSize);
            }
        }
        if (shardCount < rsConfig.DATA_SHARDS) {
            log.info("没有足够的分片~");
            return null;
        }
        for (int i = 0; i < rsConfig.TOTAL_SHARDS; i++) {
            if (!shardPresent[i]) {
                shards[i] = new byte[shardSize];
            }
        }
        rsConfig.reedSolomon.decodeMissing(shards, shardPresent, 0, shardSize);
        byte[] allBytes = new byte[shardSize * rsConfig.DATA_SHARDS];
        for (int i = 0; i < rsConfig.DATA_SHARDS; i++) {
            System.arraycopy(shards[i], 0, allBytes, shardSize * i, shardSize);
        }
        int fileSize = ByteBuffer.wrap(allBytes).getInt();

        //定位的结果不等于总数据片 修复数据
        if (ips.size() != rsConfig.TOTAL_SHARDS && rsConfig.isRepair) {
            repair(ips, metaDataInfo, shards, shardPresent);
        }
        System.out.println(allBytes.length + "------------");
        return new ByteArrayInputStream(allBytes, rsConfig.BYTES_IN_INT, fileSize);
    }

    /**
     * 修复数据片
     * @param ips
     * @param metaDataInfo
     * @param shards
     * @param shardPresent
     */
    private void repair(List<LocationMsg> ips, MetaDataInfo metaDataInfo, byte[][] shards, boolean[] shardPresent) {
        threadPoolExecutor.execute(() -> {
            List<String> exclude = ips.stream().map(LocationMsg::getAddress).collect(Collectors.toList());
            List<String> repairDataServer = DataStore.chooseRandomDataServerToString(rsConfig.TOTAL_SHARDS - ips.size(), exclude);
            int cnt = 0;
            for (int i = 0; i < rsConfig.TOTAL_SHARDS; i++) {
                if (!shardPresent[i] && cnt < repairDataServer.size()) {
                    //发送第i个数据分片 到 一个 DataServer 节点。
                    String ip = repairDataServer.get(cnt);
                    URI uri = null;
                    try {
                        uri = new URI(Sys.HTTP + ip);
                    } catch (URISyntaxException e) {
                        e.printStackTrace();
                    }
                    DataSource dataSource = new DataSource(i + "", shards[i]);
                    dataService.put(uri, metaDataInfo.getHash(), dataSource);
                    cnt++;
                }
            }
        });
    }
}
